package com.lzq.mq.broker.event.spi.listener;

import com.alibaba.fastjson.JSON;
import com.lzq.mq.broker.cache.CommonCache;
import com.lzq.mq.broker.event.model.ConsumeMsgEvent;
import com.lzq.mq.broker.model.ConsumeQueueConsumeReqModel;
import com.lzq.mq.broker.rebalance.ConsumerInstance;
import com.lzq.mq.common.dto.ConsumeMsgRespDTO;
import com.lzq.mq.common.utils.CollectionUtils;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.dto.ConsumeMsgReqDTO;
import com.lzq.mq.common.dto.ConsumeMsgRespEleDTO;
import com.lzq.mq.common.enums.BrokerResponseCode;
import com.lzq.mq.common.event.Listener;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 * 消费者拉取消息监听器
 */
@Slf4j
public class ConsumeMsgListener implements Listener<ConsumeMsgEvent> {
    @Override
    public void onReceive(ConsumeMsgEvent event) {
        try {

            log.info("开始处理消费者拉取消息请求...");
            ConsumeMsgReqDTO consumeMsgReqDTO = event.getConsumeMsgReqDTO();

            // 构建返回数据
            ConsumeMsgRespDTO consumeMsgRespDTO = new ConsumeMsgRespDTO();
            consumeMsgRespDTO.setConsumeMsgRespEleDTOList(new ArrayList<>());
            consumeMsgRespDTO.setMsgId(consumeMsgReqDTO.getMsgId());

            String topic = consumeMsgReqDTO.getTopic();
            // 计算当前请求消费的id
            String currentReqId = consumeMsgReqDTO.getIp() + ":" + consumeMsgReqDTO.getPort();

            /*
             添加消费者实例信息到缓存中
             */
            ConsumerInstance consumerInstance = new ConsumerInstance();
            consumerInstance.setIp(consumeMsgReqDTO.getIp());
            consumerInstance.setPort(consumeMsgReqDTO.getPort());
            consumerInstance.setConsumerReqId(currentReqId);
            consumerInstance.setTopic(consumeMsgReqDTO.getTopic());
            consumerInstance.setConsumeGroup(consumeMsgReqDTO.getConsumeGroup());
            consumerInstance.setBatchSize(consumeMsgReqDTO.getBatchSize());
            // 判断是否已经在消费者管理缓冲池中, 如果不再需要加入
            CommonCache.getReBalanceInstancePool().addInstancePool(consumerInstance);

            // 获取消费管理map
            Map<String, List<ConsumerInstance>> consumerGroupMap =
                    CommonCache.getConsumeHoldMap().get(topic); // 获取当前topic的消费者实例

            if (consumerGroupMap == null) {
                // 当前topic没有消费者
                event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.CONSUME_MSG_RESP.getCode(),
                        JSON.toJSONBytes(consumeMsgRespDTO)));
                return;
            }
            // 获取当前topic的消费者列表
            List<ConsumerInstance> consumerInstances = consumerGroupMap.get(consumeMsgReqDTO.getConsumeGroup());
            if (CollectionUtils.isEmpty(consumerInstances)) {
                event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.CONSUME_MSG_RESP.getCode(),
                        JSON.toJSONBytes(consumeMsgRespDTO)));
                return;
            }

            for (ConsumerInstance instance : consumerInstances) {
                if (instance.getConsumerReqId().equals(currentReqId)) {     // 找到当前消费者实例信息
                    // 判断当前消费者消费的队列中是否包含目标队列
                    // 获取当前的消费者实例, 获取其绑定的消费队列, 消费其消费的每一个队列中的消息
                    log.info("{}开始处理消费者拉取消息请求, 队列id: {}", instance.getConsumerReqId() ,instance.getQueueIdSet());
                    for (Integer queueId : instance.getQueueIdSet()) {     // 遍历当前消费者消费的队列, 从队列中获取消息
                        ConsumeQueueConsumeReqModel consumeQueueConsumeReqModel = new ConsumeQueueConsumeReqModel();
                        consumeQueueConsumeReqModel.setTopic(topic);
                        consumeQueueConsumeReqModel.setQueueId(queueId);
                        // todo 这里的定义是否有些问题, 消费者订阅的每条队列都拉batchSize条
                        // todo 此外, 是否会出现越界的情况
                        consumeQueueConsumeReqModel.setBatchSize(instance.getBatchSize());
                        consumeQueueConsumeReqModel.setConsumeGroup(instance.getConsumeGroup());

                        List<byte[]> commitLogContentList = CommonCache.getConsumeQueueConsumeHandler().consume(consumeQueueConsumeReqModel);
                        ConsumeMsgRespEleDTO consumeMsgRespEleDTO = new ConsumeMsgRespEleDTO();
                        consumeMsgRespEleDTO.setQueueId(queueId);
                        consumeMsgRespEleDTO.setCommitLogContentList(commitLogContentList);
                        consumeMsgRespDTO.getConsumeMsgRespEleDTOList().add(consumeMsgRespEleDTO);
                    }
                }
            }

            // 返回数据
            byte[] respBody = JSON.toJSONBytes(consumeMsgRespDTO);
            TcpMsg tcpMsg = new TcpMsg(BrokerResponseCode.CONSUME_MSG_RESP.getCode(), respBody);
            event.getChannelHandlerContext().writeAndFlush(tcpMsg);
            log.info("处理消费者拉取消息请求完成...");
        } catch (Exception e) {
            log.error("处理消费者拉取消息请求异常...", e);
        }
    }

}
